#include "UDPServer.h"
#include "asynevent.h"
#include "ntdproto.h"
#include "ntdtime.h"


#define MOD_NUM 100
#define INACTIVE_CON_TIMEOUT 3000//3s

static void cb_func(evutil_socket_t fd, short what, void *user_data)
{	
	if( what&EV_READ ){
		(static_cast<UDPServer *>(user_data))->OnRead();
	}

	if( what&EV_WRITE ){
		(static_cast<UDPServer *>(user_data))->OnWrite();
	}
}

static void timeout_cb(evutil_socket_t fd, short event, void *user_data)
{
	(static_cast<UDPServer *>(user_data))->OnTimer();
}


UDPServer::UDPServer()
{
	m_socket = INVALID_SOCKET;
	m_ev = NULL;
	m_evbuf_input = NULL;
	m_evbuf_output = NULL;

	m_lastrcvts = 0;

	//m_bReply = false;
	//m_pktloss = 0;
	//m_seqexpt = 1;
}

UDPServer::~UDPServer()
{
	Stop();
}

bool UDPServer::Listen(int port)
{
	m_evbuf_input = evbuffer_new();
	m_evbuf_output = evbuffer_new();

	m_socket = socket( PF_INET, SOCK_DGRAM, IPPROTO_UDP );
	if ( m_socket == INVALID_SOCKET ) 
		return false;

	//int RcvBufferSize = 12*1024*1024;
	//setsockopt( m_socket, SOL_SOCKET, SO_RCVBUF,
	//	(char*)&RcvBufferSize, sizeof( RcvBufferSize ) );

	struct sockaddr_in sin;
	SetSockAddr(sin, 0, port);

	if( bind( m_socket, (sockaddr*)&sin, sizeof(sin) ) == -1 ){
		return false;
	}

	evutil_make_socket_nonblocking(m_socket);

	//don't use EV_WRITE
	m_ev = event_new(AsynEventFramework::Instance()->GetEventBase(), m_socket, EV_READ|EV_PERSIST, cb_func,
		this);

	//timeval timeout = {1, 10000};
	event_add(m_ev, NULL);

	//Timer to remove inactive conversation
	event_assign(&m_timeout, AsynEventFramework::Instance()->GetEventBase(), -1, EV_PERSIST, timeout_cb, this);

	struct timeval tv;
	evutil_timerclear(&tv);
	tv.tv_sec = 1;
	event_add(&m_timeout, &tv);


	return true;
}

void UDPServer::Stop()
{
	std::map<int16, NtdConn*>::iterator iter = m_connMap.begin();
	std::map<int16, NtdConn*>::iterator end = m_connMap.end();
	for(; iter!=end; ++iter){
		delete iter->second;
	}
	m_connMap.clear();

	event_del(m_ev);
	event_free(m_ev);
	evutil_closesocket(m_socket);
	evbuffer_free(m_evbuf_input);
	evbuffer_free(m_evbuf_output);


	m_evbuf_input = NULL;
	m_evbuf_output = NULL;
	m_socket = INVALID_SOCKET;
	m_ev = NULL;
	//m_bReply = false;
}

void UDPServer::OnTimer()
{
	std::map<int16, NtdConn*>::iterator iter = m_connMap.begin();
	std::map<int16, NtdConn*>::iterator end = m_connMap.end();
	for(; iter!=end; ){
		if( iter->second->GetLastIOTs() + INACTIVE_CON_TIMEOUT 
			< NtdTime::Instance()->GetCurrentTime_ms() )
		{
			LOG4CPLUS_DEBUG(g_logger, "Remove conversation, conv_id = "<<iter->first);
			delete iter->second;
			iter = m_connMap.erase(iter);
		}else{
			++iter;
		}		
	}
}

void UDPServer::HandleHandShake(NtdProtoHandShake * proto, void* pContext)
{
	std::map<int16, NtdConn*>::iterator iter = m_connMap.find(proto->m_convId);
	NtdConn * pConn = NULL;
	if( iter == m_connMap.end() ){
		pConn = new NtdConn(m_socket, (*(sockaddr_in*)pContext), proto->m_ini_seq, m_evbuf_output);
		m_connMap.insert(std::make_pair(proto->m_convId, pConn));

		LOG4CPLUS_DEBUG(g_logger, "add conversation, conv_id = "<<proto->m_convId);
	}else{
		pConn = iter->second;
	}

	pConn->HandleHandShake(proto);	
}

void UDPServer::HandleFin(NtdProtoFin * proto, void* pContext)
{
	std::map<int16, NtdConn*>::iterator iter = m_connMap.find(proto->m_convId);
	NtdConn * pConn = NULL;
	if( iter == m_connMap.end() ){
		LOG4CPLUS_ERROR(g_logger, "conversation not exist, conv_id = "<<proto->m_convId);
	}else{
		pConn = iter->second;
		pConn->HandleFin(proto);	
	}
}

void UDPServer::HandleProtoData(NtdProtoData * proto, void* pContext)
{
	std::map<int16, NtdConn*>::iterator iter = m_connMap.find(proto->m_convId);
	NtdConn * pConn = NULL;
	if( iter == m_connMap.end() ){
		LOG4CPLUS_ERROR(g_logger, "Conversation not found, conv_id = "<<proto->m_convId);
		return;
	}else{
		pConn = iter->second;
	}

	pConn->HandleProtoData(proto);

	//m_sm.Add(proto->GetPaketSize());

	////LOG4CPLUS_DEBUG(g_logger," [server] - "<< proto->m_ptk_seq
	////	<< " PKT_SIZE = "<<proto->GetPaketSize());

	////pkt loss
	//if( proto->m_ptk_seq > m_seqexpt ){
	//	m_pktloss += (proto->m_ptk_seq - m_seqexpt);
	//}else if( proto->m_ptk_seq < m_seqexpt ){
	//	LOG4CPLUS_ERROR(g_logger, "disorder pkt, seq = "<<proto->m_ptk_seq
	//		<<" expect = "<< m_seqexpt);
	//	//assert(false);
	//}

	//m_seqexpt = proto->m_ptk_seq + 1;

	////if( proto->m_ptk_seq % MOD_NUM == 0)
	//{
	//	m_feedback.m_pkt_loss = m_pktloss*100/(m_seqexpt-1);
	//	m_feedback.m_recving_rate = m_sm.GetSpeed();
	//	m_feedback.m_timestamp_sending = proto->m_timestamp_sending;
	//	m_feedback.m_timestamp_recving = NtdTime::Instance()->GetCurrentTime_ms();

	//	m_feedback.Encode(m_evbuf_output);

	//	m_bReply = true;

	//	LOG4CPLUS_TRACE(g_logger," [server] - "<<m_feedback.Desc());
	//}

}

void UDPServer::OnRead()
{
	int8 packet[1500];
	struct sockaddr_in addr;
	ev_socklen_t addrlen;
	int r;

	while(true) {
		addrlen = sizeof(struct sockaddr_in);
		//if(m_lastrcvts != 0){
		//	LOG4CPLUS_DEBUG(g_logger, "recv ts: "<<Now_us()-m_lastrcvts);
		//} 
		//m_lastrcvts = Now_us();
		r = recvfrom(m_socket, packet, sizeof(packet), 0, (struct sockaddr*) &addr, &addrlen);
		LOG4CPLUS_TRACE(g_logger, "on read, recv ts: "<<Now_us()<<" len = "<<r);
		if (r < 0) {
			int err = evutil_socket_geterror(m_socket);
			if( !UTDUTIL_ERR_RW_RETRIABLE(err) ){
				LOG4CPLUS_ERROR(g_logger,
					"Error while reading, "<<evutil_socket_error_to_string(err));
			}
			return;
		}

		if( r == 0 ){
			LOG4CPLUS_ERROR(g_logger,"Incorrect UDP packet");
			return;
		}
		
		evbuffer_add(m_evbuf_input, packet, r);

		bool ret = NtdProtoParser::ParseProtocol(m_evbuf_input, &addr);
		if( !ret || evbuffer_get_length(m_evbuf_input) != 0 ){
			LOG4CPLUS_ERROR(g_logger, "Error length while reading, ");
			evbuffer_drain(m_evbuf_input, evbuffer_get_length(m_evbuf_input));
			return;
		}

		//if( m_bReply ){
		//	//send packets
		//	int32 length = evbuffer_get_length(m_evbuf_output);
		//	evbuffer_remove(m_evbuf_output, packet, length);

		//	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&addr, sizeof(addr));
		//    m_bReply = false;
		//}
	}

}

void UDPServer::OnWrite()
{
}


NtdConn::NtdConn(evutil_socket_t s, const sockaddr_in& addr, int32 seq, evbuffer * output)
{
	m_socket = s;
	m_initSeq = seq;
	memcpy(&m_addr, &addr, sizeof(addr));
	m_output = output;
	m_pktloss = 0;
	m_seqexpt = m_initSeq+1;
	m_lastIOts = NtdTime::Instance()->GetCurrentTime_ms();

	//m_tsstart = m_lastIOts;//should be after rece first data
	m_tscost = 1;
	m_total = 0;

	m_bPacketPair = false;
	m_tsFirst = 0;
	m_bFin = false;
}

NtdConn::~NtdConn()
{
	//bool bInc = inc_dect.CalcIncreasingTrend();

	//LOG4CPLUS_DEBUG(g_logger, "Increasing trend: "<<bInc<<" total ="<<byte_format(m_total, 'A')
	//	<<" cost = "<<m_tscost
	//	<<" rcv_rate = "<<byte_format(m_total*1000.0/m_tscost, 'A')
	//	<<"/s loss : ("<<m_pktloss<<"/"<<m_seqexpt-1<<")");
}

void NtdConn::HandleProtoData(NtdProtoData * proto)
{
	if( m_bFin ) return;

	uint64 ts = NtdTime::Instance()->GetCurrentTime_us();

	//process packet pair
	if(m_bPacketPair){
		LOG4CPLUS_TRACE(g_logger,"rcv packet, ts = "<<ts);
		if( proto->m_ptk_seq%2 == 0 ){
			//first packet
			m_tsFirst = ts;
		}else if( proto->m_ptk_seq == m_seqexpt ){
			if( ts != m_tsFirst ){
				uint32 estimatebw = proto->GetPaketSize()*1000000/(ts - m_tsFirst);
				LOG4CPLUS_DEBUG(g_logger, "packet pair estimate bw: "<< byte_format(estimatebw, 'A')<<"/s" << " ts = "<<(ts - m_tsFirst));
			}else{
				assert(false);
			}
		}
	}

	m_lastIOts = ts/1000;
	if( m_total == 0 ){
		m_tsstart = m_lastIOts;
	}

	//m_sm.Add(proto->GetPaketSize());
	m_total += proto->GetPaketSize();
	m_tscost = m_lastIOts - m_tsstart;

	LOG4CPLUS_TRACE(g_logger," [server] - "<< proto->Desc());

	//pkt loss
	if( NtdSeq::seqoff(m_seqexpt, proto->m_ptk_seq) > 0 ){
		m_pktloss += NtdSeq::seqoff(m_seqexpt, proto->m_ptk_seq);
		LOG4CPLUS_ERROR(g_logger,"pkt loss, "<<NtdSeq::seqoff(m_seqexpt, proto->m_ptk_seq)<< " total_loss = "<<m_pktloss);
		m_seqexpt = NtdSeq::incseq(proto->m_ptk_seq);
	}else if( NtdSeq::seqoff(m_seqexpt, proto->m_ptk_seq) == 0){
		m_seqexpt = NtdSeq::incseq(proto->m_ptk_seq);
	}else{
		LOG4CPLUS_ERROR(g_logger, "disorder pkt, seq = "<<proto->m_ptk_seq
			<<" expect = "<< m_seqexpt);
		if( m_pktloss > 0)
			--m_pktloss;
		//assert(false);
	}

	//m_seqexpt = NtdSeq::incseq(proto->m_ptk_seq);//evil

	//if( proto->m_ptk_seq % MOD_NUM == 0)
	if(m_tsstart != m_lastIOts)//skip first packet
	{
		assert(m_tscost);
		assert(m_seqexpt!=m_initSeq);
		m_feedback.m_pkt_loss = m_pktloss*100/NtdSeq::seqoff(m_initSeq, m_seqexpt);
		//m_feedback.m_recving_rate = m_sm.GetSpeed();
		m_feedback.m_recving_rate = static_cast<uint32>(m_total*1000/m_tscost);
		m_feedback.m_timestamp_sending = proto->m_timestamp_sending;
		m_feedback.m_timestamp_recving = ts/1000;

		m_feedback.m_convId = proto->m_convId;


		LOG4CPLUS_TRACE(g_logger," [server] - "<<m_feedback.Desc());
		LOG4CPLUS_TRACE(g_logger,"recv: "<<m_feedback.m_timestamp_recving<< " send: "<<m_feedback.m_timestamp_sending
			<<" sample: "<<static_cast<uint64>(m_feedback.m_timestamp_recving - m_feedback.m_timestamp_sending));

		inc_dect.AddSample(m_feedback.m_timestamp_recving - m_feedback.m_timestamp_sending);

		//send packets
		//m_feedback.Encode(m_output);
		//int32 length = evbuffer_get_length(m_output);
		//int8 packet[1500];
		//evbuffer_remove(m_output, packet, length);

		//int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_addr, sizeof(m_addr));
	}
}

void NtdConn::HandleHandShake(NtdProtoHandShake * proto)
{
	NtdProtoHandShakeResp resp;
	resp.m_convId = proto->m_convId;
	resp.m_ini_seq = proto->m_ini_seq;
	resp.m_load = 0;
	m_bPacketPair = proto->m_bPacketPair;

	resp.Encode(m_output);

	LOG4CPLUS_TRACE(g_logger," [server] - "<<resp.Desc());

	//send packets
	int32 length = evbuffer_get_length(m_output);
	int8 packet[1500];
	evbuffer_remove(m_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_addr, sizeof(m_addr));
	
	m_lastIOts = NtdTime::Instance()->GetCurrentTime_ms();
}

void NtdConn::HandleFin(NtdProtoFin * proto)
{
	m_bFin = true;

	bool bInc = inc_dect.CalcIncreasingTrend();
	assert(m_seqexpt != m_initSeq);

	LOG4CPLUS_DEBUG(g_logger, "Increasing trend: "<<bInc<<" total ="<<byte_format(m_total, 'A')
		<<" cost = "<<m_tscost
		<<" rcv_rate = "<<byte_format(m_total*1000.0/m_tscost, 'A')
		<<"/s loss : ("<<m_pktloss<<"/"<<NtdSeq::seqoff(m_initSeq, m_seqexpt)<<")");

	NtdProtoFinResp resp;
	resp.m_convId = proto->m_convId;
	resp.m_recv_rate = static_cast<uint32>(m_total*1000.0/m_tscost);
	resp.m_bCongestion = bInc;
	resp.m_pkt_loss = m_pktloss*100/NtdSeq::seqoff(m_initSeq, m_seqexpt);

	resp.Encode(m_output);

	LOG4CPLUS_TRACE(g_logger," [server] - "<<resp.Desc());

	//send packets
	int32 length = evbuffer_get_length(m_output);
	int8 packet[1500];
	evbuffer_remove(m_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_addr, sizeof(m_addr));

	m_lastIOts = NtdTime::Instance()->GetCurrentTime_ms();
}
